package org.hbhk.hxl.gateway.route;

import static org.springframework.cloud.gateway.filter.headers.HttpHeadersFilter.filterRequest;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.CLIENT_RESPONSE_ATTR;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.CLIENT_RESPONSE_CONN_ATTR;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.CLIENT_RESPONSE_HEADER_NAMES;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.PRESERVE_HOST_HEADER_ATTRIBUTE;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.isAlreadyRouted;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.setAlreadyRouted;

import java.net.URI;
import java.time.Duration;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.hbhk.hxl.gateway.exception.RouteException;
import org.hbhk.hxl.gateway.filter.GatewayFilter;
import org.hbhk.hxl.gateway.model.HttpService;
import org.hbhk.hxl.gateway.services.ServiceUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.cloud.gateway.config.HttpClientProperties;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.cloud.gateway.filter.headers.HttpHeadersFilter;
import org.springframework.cloud.gateway.filter.headers.HttpHeadersFilter.Type;
import org.springframework.cloud.gateway.support.TimeoutException;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.NettyDataBuffer;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.AbstractServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.server.ResponseStatusException;
import org.springframework.web.server.ServerWebExchange;

import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.HttpClientResponse;

public class TimeoutNettyRoutingFilter implements GlobalFilter, Ordered {

	private final HttpClient httpClient;
	private final ObjectProvider<List<HttpHeadersFilter>> headersFilters;
	private final HttpClientProperties properties;

	private String startGray = "0";

	public final static String GATEWAY_REQUEST_URL_TIMEOUT = "gateway_request_url_timeout";

	private final Logger logger = LoggerFactory.getLogger(TimeoutNettyRoutingFilter.class);

	public TimeoutNettyRoutingFilter(HttpClient httpClient, ObjectProvider<List<HttpHeadersFilter>> headersFilters,
			HttpClientProperties properties) {
		this.httpClient = httpClient;
		this.headersFilters = headersFilters;
		this.properties = properties;
	}

	@Override
	public int getOrder() {
		return Ordered.LOWEST_PRECEDENCE;
	}

	@Override
	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
		long s = System.currentTimeMillis();
		URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);
		String scheme = requestUrl.getScheme();
		if (isAlreadyRouted(exchange) || (!"http".equals(scheme) && !"https".equals(scheme))) {
			return chain.filter(exchange);
		}
		setAlreadyRouted(exchange);

		ServerHttpRequest request = exchange.getRequest();

		final HttpMethod method = HttpMethod.valueOf(request.getMethodValue());
		String url = requestUrl.toString();
		String urlPath = exchange.getRequiredAttribute(GatewayFilter.GATEWAY_REQUEST_ROUTE_URL);
		String surl = exchange.getRequiredAttribute(GatewayFilter.GATEWAY_REQUEST_URL_ATTR_SURL);
		String gray = "0";
		if ("1".equals(startGray)) {
			HttpCookie cookie = request.getCookies().getFirst("gray");
			if (cookie != null) {
				gray = cookie.getValue();
			}
			if (StringUtils.isEmpty(gray)) {
				gray = request.getHeaders().getFirst("gray");
			}
		}
		List<String> nodes = ServiceUtil.getNode(surl, gray);
		if (nodes == null || nodes.size() == 0) {
			throw new RouteException("服务节点不存在:" + urlPath);
		}

		long s1 = System.currentTimeMillis();

		logger.debug("组装url耗时:" + (s1 - s));

		String addr = MsNodeBalanceRule.roundRobin(nodes, surl);

		String strParam = "";
		if (url.contains("?")) {
			strParam = url.substring(url.indexOf("?"), +url.length());
		}

		url = scheme + "://" + addr + urlPath + strParam;

		exchange.getAttributes().put(GatewayFilter.GATEWAY_REQUEST_FULL_ROUTE_URL, url);

		// String urlPath =
		// exchange.getRequiredAttribute(GatewayFilter.GATEWAY_REQUEST_URL_ATTR_SURL);
//		final String url = requestUrl.getScheme()+"://"+requestUrl.getHost()+":"+requestUrl.getPort()+urlPath;

		HttpHeaders filtered = filterRequest(this.headersFilters.getIfAvailable(), exchange);

		final DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders();
		filtered.forEach(httpHeaders::set);

		// String transferEncoding =
		// request.getHeaders().getFirst(HttpHeaders.TRANSFER_ENCODING);
		// boolean chunkedTransfer = "chunked".equalsIgnoreCase(transferEncoding);

		boolean preserveHost = exchange.getAttributeOrDefault(PRESERVE_HOST_HEADER_ATTRIBUTE, false);
		// httpClient.request(method).u

		Flux<HttpClientResponse> responseFlux = this.httpClient.request(method).uri(url).send((req, nettyOutbound) -> {
			req.headers(httpHeaders);

			if (preserveHost) {
				String host = request.getHeaders().getFirst(HttpHeaders.HOST);
				req.header(HttpHeaders.HOST, host);
			}
			return nettyOutbound
					.send(request.getBody().map(dataBuffer -> ((NettyDataBuffer) dataBuffer).getNativeBuffer()));
		}).responseConnection((res, connection) -> {
			ServerHttpResponse response = exchange.getResponse();
			// put headers and status so filters can modify the response
			HttpHeaders headers = new HttpHeaders();

			res.responseHeaders().forEach(entry -> headers.add(entry.getKey(), entry.getValue()));

			String contentTypeValue = headers.getFirst(HttpHeaders.CONTENT_TYPE);
			if (StringUtils.isNotEmpty(contentTypeValue)) {
				if (contentTypeValue.trim().indexOf("json") > -1) {
					if (headers.get("Location") != null) {
						headers.remove("Location");
					}
				}
				exchange.getAttributes().put(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR, contentTypeValue);
			}

			HttpStatus status = HttpStatus.resolve(res.status().code());
			if (status != null) {
				response.setStatusCode(status);
			} else if (response instanceof AbstractServerHttpResponse) {
				// https://jira.spring.io/browse/SPR-16748
				((AbstractServerHttpResponse) response).setRawStatusCode(res.status().code());
			} else {
				throw new IllegalStateException(
						"Unable to set status code on response: " + res.status().code() + ", " + response.getClass());
			}

			// make sure headers filters run after setting status so it is available in
			// response
			HttpHeaders filteredResponseHeaders = HttpHeadersFilter.filter(this.headersFilters.getIfAvailable(),
					headers, exchange, Type.RESPONSE);

			if (!filteredResponseHeaders.containsKey(HttpHeaders.TRANSFER_ENCODING)
					&& filteredResponseHeaders.containsKey(HttpHeaders.CONTENT_LENGTH)) {
				// It is not valid to have both the transfer-encoding header and the
				// content-length header
				// remove the transfer-encoding header in the response if the content-length
				// header is presen
				response.getHeaders().remove(HttpHeaders.TRANSFER_ENCODING);
			}

			exchange.getAttributes().put(CLIENT_RESPONSE_HEADER_NAMES, filteredResponseHeaders.keySet());

			response.getHeaders().putAll(filteredResponseHeaders);

			// Defer committing the response until all route filters have run
			// Put client response as ServerWebExchange attribute and write response later
			// NettyWriteResponseFilter
			exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
			exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);

			return Mono.just(res);
		});

		Duration responseTimeout = null;
		HttpService httpService = ServiceUtil.getHttpService(surl);

		if (httpService != null && httpService.getTimeout() != null) {
			responseTimeout = Duration.ofMillis(httpService.getTimeout());
		}
		if (responseTimeout == null) {
			responseTimeout = properties.getResponseTimeout();
		}
		if (responseTimeout != null) {
			responseFlux = responseFlux
					.timeout(responseTimeout,
							Mono.error(new TimeoutException(
									"Response took longer than timeout: " + properties.getResponseTimeout())))
					.onErrorMap(TimeoutException.class,
							th -> new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, null, th));
		}

		Mono<Void> mono = responseFlux.then(chain.filter(exchange));
		long s2 = System.currentTimeMillis();

		logger.debug("调用url耗时:" + (s2 - s1));
		return mono;
	}

	public String getStartGray() {
		return startGray;
	}

	public void setStartGray(String startGray) {
		this.startGray = startGray;
	}

}